//! table_cache implements a cache providing access to the immutable SSTables on disk. It's a
//! read-through cache, meaning that non-present tables are read from disk and cached before being
//! returned.

use crate::block::Block;
use crate::cache::{self, Cache};
use crate::error::{err, Result, StatusCode};
use crate::key_types::InternalKey;
use crate::options::Options;
use crate::table_reader::Table;
use crate::types::{FileNum, Shared};

use integer_encoding::FixedIntWriter;

use bytes::Bytes;
use std::convert::AsRef;
use std::path::{Path, PathBuf};
use std::rc::Rc;

/// 传一个 path_name, 后缀加上 ldb 
pub fn table_file_name<P: AsRef<Path>>(name: P, num: FileNum) -> PathBuf {
    assert!(num > 0);
    name.as_ref().join(format!("{:06}.ldb", num))
}

/// 这个 filenum_to_key 方法用于将 文件编号（FileNum）转换为缓存键（cache::CacheKey），以便在缓存系统中标识和查找与该文件相关的数据。
fn filenum_to_key(num: FileNum) -> cache::CacheKey {
    let mut buf = [0; 16];
    // 通过 write_fixedint 方法将文件编号 num 以固定长度的二进制格式写入 buf 的前几个字节（例如 u64 会占用 8 字节）。
    (&mut buf[..]).write_fixedint(num).unwrap();
    buf
}

/// TableCache 结构体是数据库中用于缓存 SSTable 文件及其数据块的核心组件，主要作用是减少磁盘 I/O 操作，提升读取性能。
/// 它整合了文件级缓存和数据块级缓存，是连接内存与磁盘数据的关键中间层。
pub struct TableCache {
    /// 数据库的文件系统路径，用于定位磁盘上的 SSTable 文件（当缓存未命中时，需要从该路径加载文件）。
    dbname: PathBuf,
    /// SSTable 文件对象的缓存（Table 是 SSTable 文件在内存中的表示，包含文件的元数据和索引）。
    /// 缓存键通常是文件编号（通过 filenum_to_key 转换），值是加载后的 Table 对象，避免频繁打开 / 解析磁盘上的 SSTable 文件。
    cache: Cache<Table>,
    /// SSTable 数据块（Block）的共享缓存。
    /// 数据块是 SSTable 中实际存储键值对的数据单元，缓存常用的数据块可以大幅减少磁盘读取次数（例如，多次查询同一范围内的键时，直接从内存缓存获取数据块）。
    block_cache: Shared<Cache<Block>>,
    /// 数据库配置选项，可能包含缓存大小、压缩方式等参数，用于控制缓存行为（如缓存淘汰策略、数据块解压方式等）。
    opts: Options,
}

impl TableCache {
    /// Create a new TableCache for the database named `db`, caching up to `entries` tables.
    ///
    /// opt.cmp should be the user-supplied comparator.
    pub fn new<P: AsRef<Path>>(
        db: P,
        opt: Options,
        block_cache: Shared<Cache<Block>>,
        entries: usize,
    ) -> TableCache {
        TableCache {
            dbname: db.as_ref().to_owned(),
            cache: Cache::new(entries),
            block_cache,
            opts: opt,
        }
    }

    /// 这个 get 方法是管理多个 SSTable 文件的容器（通常是 LSM 树中的某个层级，
    /// 如 Level 或 Version）提供的查询接口，用于根据文件编号（file_num）和内部键（InternalKey）从指定的 SSTable 中查找对应的值。
    /// 它是多层存储结构中定位特定文件并查询数据的中间层接口。
    ///
    /// 这个 get 方法的核心价值在于实现 “按文件编号定位 SSTable 并查询数据” 的逻辑封装，主要作用包括：
    ///
    /// 解耦文件管理与查询逻辑
    ///
    /// 上层调用无需关心 SSTable 如何存储、加载和缓存，只需提供文件编号和查询键，即可获取结果。
    ///
    /// 利用缓存提升效率
    ///
    /// 内部通过 get_table 可能从缓存中获取 SSTable 实例（如 table_cache），避免重复打开磁盘文件，减少 I/O 开销。
    ///
    /// 统一查询接口
    ///
    /// 为多层级的 SSTable 集合（如 LSM 树的不同 Level）提供一致的查询入口，简化上层存储结构的复杂性。
    pub fn get(
        &mut self,
        file_num: FileNum,
        key: InternalKey<'_>,
    ) -> Result<Option<(Bytes, Bytes)>> {
        // 调用 self.get_table(file_num) 根据文件编号查找对应的 SSTable 实例（tbl）；
        // 这里的 get_table 通常会从缓存或磁盘加载 SSTable（例如从 table_cache 中获取，避免重复打开文件）；
        // 若文件不存在或加载失败，返回 Err 错误。
        let tbl = self.get_table(file_num)?;
        // 调用 SSTable 自身的 get 方法（如之前分析过的 SSTable::get），传入内部键 key 执行实际查询；
        // 返回查询结果：Some((key, value)) 表示找到键值对，None 表示未找到，或封装错误信息的 Err。
        // 这里 BlockIter 进
        tbl.get(key)
    }

    /// Return a table from cache, or open the backing file, then cache and return it.
    ///
    /// 这个 get_table 方法是SSTable 缓存管理的核心接口，负责从缓存中获取指定的 SSTable 实例；如果缓存未命中，则打开对应的磁盘文件并加载 SSTable，
    /// 同时将其存入缓存以便后续复用。它通过缓存机制减少重复打开文件和解析 SSTable 的开销，显著提升数据库的查询性能。
    pub fn get_table(&mut self, file_num: FileNum) -> Result<Table> {
        // 通过 filenum_to_key 将文件编号（file_num）转换为缓存键（key），用于在缓存中标识该 SSTable。
        let key = filenum_to_key(file_num);
        // 尝试从缓存（self.cache）中获取该键对应的 SSTable 实例（t）；
        // 若缓存命中（Some(t)），直接返回该实例的副本（通常是轻量的 Rc 或 Arc 克隆，不复制实际数据），避免重复加载。
        if let Some(t) = self.cache.get(&key) {
            return Ok(t.clone());
        }
        // 若缓存未命中，调用 open_table 方法从磁盘打开对应的 SSTable 文件：
        // 读取文件元数据（如索引块、过滤器等）；
        // 解析并创建 Table 实例；
        // 通常 open_table 内部会将新创建的 Table 实例存入缓存（self.cache.insert(...)），供后续查询复用；
        // 返回打开的 Table 实例（或错误）。
        self.open_table(file_num)
    }

    /// Open a table on the file system and read it.
    ///
    /// 这个 open_table 方法是从磁盘加载 SSTable 文件并创建对应内存实例的核心函数。
    /// 它负责处理文件路径解析、文件大小校验、文件打开以及最终的 SSTable 实例化，并将加载后的实例存入缓存，为后续查询提供支持。
    ///
    /// open_table 方法的核心价值在于实现 SSTable 从磁盘文件到内存可用实例的完整加载流程，具体包括：
    ///
    /// 文件验证：确保加载的是有效、非空的 SSTable 文件；
    ///
    /// 资源管理：通过 Rc 共享文件句柄，优化系统资源使用；
    ///
    /// 实例化：解析 SSTable 内部结构，创建可直接用于查询的 Table 实例；
    ///
    /// 缓存维护：加载后自动存入缓存，避免后续重复加载的开销。
    fn open_table(&mut self, file_num: FileNum) -> Result<Table> {
        // 通过 table_file_name 函数，结合数据库目录（self.dbname）和文件编号（file_num），生成 SSTable 在磁盘上的完整文件名（例如 000003.sst）；
        // 将文件名转换为文件系统路径（Path），用于后续文件操作。
        let name = table_file_name(&self.dbname, file_num);
        let path = Path::new(&name);
        // 调用文件系统接口（self.opts.env.size_of）获取文件大小；
        // 若文件大小为 0，返回 “无效数据” 错误（空文件不可能是有效的 SSTable）。
        let file_size = self.opts.env.size_of(path)?;
        if file_size == 0 {
            return err(StatusCode::InvalidData, "file is empty");
        }
        // 通过环境接口（open_random_access_file）打开文件，获取随机访问句柄（支持按偏移量读取，无需顺序访问）；
        // 用 Rc 包裹文件句柄，实现多个 Table 实例共享同一文件句柄（减少系统资源占用）。
        let file = Rc::new(self.opts.env.open_random_access_file(path)?);
        // No SSTable file name compatibility.
        // 调用 Table::new 解析文件内容，创建内存中的 Table 实例：
        // 读取 SSTable 的元数据（如索引块、布隆过滤器、 Footer 等）；
        // 关联块缓存（block_cache），用于缓存数据块；
        // 验证文件格式的合法性（如校验 Footer 中的魔术数）。
        let table = Table::new(self.opts.clone(), self.block_cache.clone(), file, file_size)?;
        // 将新创建的 Table 实例存入缓存（self.cache），键为文件编号生成的缓存键；
        // 返回 Table 实例，供上层查询使用。
        self.cache.insert(&filenum_to_key(file_num), table.clone());
        Ok(table)
    }

    /// 从cache里面删除该文件
    pub fn evict(&mut self, file_num: FileNum) -> Result<()> {
        if self.cache.remove(&filenum_to_key(file_num)).is_some() {
            Ok(())
        } else {
            err(StatusCode::NotFound, "table not present in cache")
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::cache;
    use crate::mem_env::MemEnv;
    use crate::options;
    use crate::table_builder::TableBuilder;
    use crate::test_util::LdbIteratorIter;
    use crate::types::share;

    #[test]
    fn test_table_file_name() {
        assert_eq!(Path::new("abc/000122.ldb"), table_file_name("abc", 122));
        assert_eq!(
            Path::new("abc/1234567.ldb"),
            table_file_name("abc", 1234567)
        );
    }

    fn make_key(a: u8, b: u8, c: u8) -> cache::CacheKey {
        [a, b, c, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
    }

    #[test]
    fn test_filenum_to_key() {
        assert_eq!(make_key(16, 0, 0), filenum_to_key(0x10));
        assert_eq!(make_key(16, 1, 0), filenum_to_key(0x0110));
        assert_eq!(make_key(1, 2, 3), filenum_to_key(0x030201));
    }

    fn write_table_to(o: Options, p: &Path) {
        let w = o.env.open_writable_file(p).unwrap();
        let mut b = TableBuilder::new_raw(o, w);

        let data = [
            ("abc", "def"),
            ("abd", "dee"),
            ("bcd", "asa"),
            ("bsr", "a00"),
        ];

        for &(k, v) in data.iter() {
            b.add(k.as_bytes(), v.as_bytes()).unwrap();
        }
        b.finish().unwrap();
    }

    #[test]
    fn test_table_cache() {
        // Tests that a table can be written to a MemFS file, read back by the table cache and
        // parsed/iterated by the table reader.
        let mut opt = options::for_test();
        opt.env = Rc::new(Box::new(MemEnv::new()));
        let bc = share(Cache::new(128));
        let dbname = Path::new("testdb1");
        let tablename = table_file_name(dbname, 123);
        let tblpath = Path::new(&tablename);

        write_table_to(opt.clone(), tblpath);
        assert!(opt.env.exists(tblpath).unwrap());
        assert!(opt.env.size_of(tblpath).unwrap() > 20);

        let mut cache = TableCache::new(dbname, opt.clone(), bc, 10);
        assert!(cache.cache.get(&filenum_to_key(123)).is_none());
        assert_eq!(
            LdbIteratorIter::wrap(&mut cache.get_table(123).unwrap().iter()).count(),
            4
        );
        // Test cached table.
        assert_eq!(
            LdbIteratorIter::wrap(&mut cache.get_table(123).unwrap().iter()).count(),
            4
        );

        assert!(cache.cache.get(&filenum_to_key(123)).is_some());
        assert!(cache.evict(123).is_ok());
        assert!(cache.evict(123).is_err());
        assert!(cache.cache.get(&filenum_to_key(123)).is_none());
    }
}
